package com.kyp.basic;

import java.io.IOException;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.log4j.Logger;

import com.kyp.util.Validation;

public class BasicCalcMapper extends
		Mapper<LongWritable, Text, BasicCalcKey, DoubleWritable> {

	private static final Logger sLogger = Logger
			.getLogger(BasicCalcMapper.class);

	private int mGroupCount;

	private BasicCalcKey mOutputKey = new BasicCalcKey();

	private DoubleWritable mOutputValue = new DoubleWritable();

	@Override
	protected void setup(Context context) throws IOException,
			InterruptedException {
		sLogger.info("***************Mapper Start***************");
		
		// Left Group Column Count
		mGroupCount = Integer.parseInt(context.getConfiguration().get("group"));
	}

	@Override
	protected void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		
		if (key.get() > 0) {
			String[] columns = value.toString().split(",");
			if (columns != null && columns.length > 0) {

				// Left Group 과 나머지 Column Index가 Key
				for (int grpi = 0; grpi < mGroupCount; grpi++) {
					try {
						for (int coli = mGroupCount; coli < columns.length; coli++) {
							mOutputKey.setGroupColumnKey(columns[grpi]
									.toString());
							mOutputKey.setColumnIndex(coli);

							if (Validation.isNumeric(columns[coli].toString()))
								mOutputValue.set(Double
										.parseDouble(columns[coli].toString()));
							else
								mOutputValue.set(0);

							context.write(mOutputKey, mOutputValue);
						}

					} catch (Exception e) {
						sLogger.error(e.getMessage(), e);
					}
				}
			}
		}
	}
}
